package com.ideal.frame.node.netty.job;

import com.ideal.frame.node.entity.NodeTb;
import com.ideal.frame.node.entity.RootNodeTb;
import com.ideal.frame.node.enumeration.NodeLevel;
import com.ideal.frame.node.netty.NettyClient;
import com.ideal.frame.node.netty.client.NettyClientConstant;
import com.ideal.frame.node.service.NodeService;
import com.ideal.frame.node.service.RootNodeService;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @description:
 * @author: xuyang
 * @date: 2021/9/2 13:48
 */
@Component
@Slf4j
public class HeartbeatScheduler {

    @Autowired
    NodeService nodeService;

    @Autowired
    RootNodeService rootNodeService;

    @Autowired
    NettyClient nettyClient;

    /**
     * 默认每分钟查询一次数据库，取出不心跳检测的节点列表
     */
    @Scheduled(cron = "${netty.client.cron.cancel-heartbeat: 0 */1 * * * ?}")
    public void getCancelHeartbeatNodes(){
        NettyClientConstant.cancelHeartbeatList = new ArrayList<>();
        List<NodeTb> list = nodeService.getAllCancelHeartbeatNodes();
        List<RootNodeTb> allCancelHeartbeatNodes = rootNodeService.getAllCancelHeartbeatNodes();
        log.info("**********************查询已经取消心跳检测的记录***********************");
        log.info("取消心跳检测共{}条，host和ip是{}", list.size(), list.stream().map(item->item.getNodeip()+":"+item.getNettyport()).collect(Collectors.toList()));
        log.info("根节点取消心跳检测共{}条，host和ip是{}", allCancelHeartbeatNodes.size(), allCancelHeartbeatNodes.stream().map(item->item.getNodeip()+":"+item.getNettyport()).collect(Collectors.toList()));
        for(NodeTb node : list){
            String hostAndPort = node.getNodeip() + ":" + node.getNettyport();
            NettyClientConstant.cancelHeartbeatList.add(hostAndPort);
            // 取消心跳检测后，清除连接次数
            NettyClientConstant.retryConnTimesMap.remove(hostAndPort);
            // 关闭心跳检测通道
            if(NettyClientConstant.channelMaps.get(hostAndPort) != null){
                Channel channel = NettyClientConstant.channelMaps.get(hostAndPort).getChannel();
                if(channel != null){
                    channel.close();
                }
            }
        }
        for(RootNodeTb node : allCancelHeartbeatNodes){
            String hostAndPort = node.getNodeip() + ":" + node.getNettyport();
            NettyClientConstant.cancelHeartbeatList.add(hostAndPort);
            // 取消心跳检测后，清除连接次数
            NettyClientConstant.retryConnTimesMap.remove(hostAndPort);
            // 关闭心跳检测通道
            if(NettyClientConstant.channelMaps.get(hostAndPort) != null){
                Channel channel = NettyClientConstant.channelMaps.get(hostAndPort).getChannel();
                if(channel != null){
                    channel.close();
                }
            }
        }
        // 取出所有节点并过滤，去除nettyMonitor为false的记录，去除已连接的记录
        List<NodeTb> allList = nodeService.getAllNodes();
        List<NodeTb> remainedList = allList.stream()
                // 心跳检测状态为true及激活状态，否则过滤掉
                .filter(item->item.getNettymonitor())
                // 不能是当前自己本节点，只能监测下级节点
                .filter(item-> !item.getNodelevel().equals(NettyClientConstant.selfNodeLevel))
                // 不能是已经处于连接状态的节点
                .filter(item -> !NettyClientConstant.cancelHeartbeatList.contains(item.getNodeip()+":"+item.getNettyport()))
                .filter(item -> !NettyClientConstant.channelMaps.containsKey(item.getNodeip()+":"+item.getNettyport()))
                // 不能是已经在重连的记录
                .filter(item -> !NettyClientConstant.retryConnTimesMap.containsKey(item.getNodeip()+":"+item.getNettyport()))
                .collect(Collectors.toList());
        // 剩余的节点重新进行心跳检测
        log.info("新增心跳检测共{}条，host和ip是{}", remainedList.size(), remainedList.stream().map(item->item.getNodeip()+":"+item.getNettyport()).collect(Collectors.toList()));
        for (NodeTb node : remainedList){
            nettyClient.doConncet(node.getNodeip(), node.getNettyport(), node.getId(), node.getNodelevel());
        }
        List<RootNodeTb> allRootList = rootNodeService.getAllNodes();
        List<RootNodeTb> remainedRootList = allRootList.stream()
                // 心跳检测状态为true及激活状态，否则过滤掉
                .filter(item->item.getNettymonitor())
                // 不能是当前自己本节点，只能监测下级节点
//                .filter(item-> !item.getNodelevel().equals(NettyClientConstant.selfNodeLevel))
                // 不能是已经处于连接状态的节点
                .filter(item -> !NettyClientConstant.cancelHeartbeatList.contains(item.getNodeip()+":"+item.getNettyport()))
                .filter(item -> !NettyClientConstant.channelMaps.containsKey(item.getNodeip()+":"+item.getNettyport()))
                // 不能是已经在重连的记录
                .filter(item -> !NettyClientConstant.retryConnTimesMap.containsKey(item.getNodeip()+":"+item.getNettyport()))
                .collect(Collectors.toList());
        // 剩余的节点重新进行心跳检测
        log.info("新增跟节点（节点登记-1）心跳检测共{}条，host和ip是{}", remainedRootList.size(), remainedRootList.stream().map(item->item.getNodeip()+":"+item.getNettyport()).collect(Collectors.toList()));
        for (RootNodeTb node : remainedRootList){
            nettyClient.doConncet(node.getNodeip(), node.getNettyport(), node.getId(), NodeLevel.CHILD_ROOT.getValue());
        }
    }
}
